package com.cx.zyx.modulebiz.news.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.cx.zyx.modulebiz.news.mapper.NewsMapper;
import com.cx.zyx.modulebiz.news.model.News;
import com.cx.zyx.modulebiz.news.service.NewsService;
import com.cx.zyx.modulebiz.news.service.NewsStatsService;
import com.cx.zyx.modulebiz.news.service.dto.TopNewsDTO;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.aop.framework.AopContext;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author zjg
 *  针对表【news】的数据库操作Service实现
 * @since  2025-07-14 14:50:23
 */
@Slf4j
@Service
@EnableAspectJAutoProxy(exposeProxy = true)  // 启用 AopContext
public class NewsServiceImpl extends ServiceImpl<NewsMapper, News> implements NewsService {

    private final RocketMQTemplate rocketMQTemplate;

    private final NewsStatsService newsStatsService;

    // 注入RocketMQTemplate
    public NewsServiceImpl(RocketMQTemplate rocketMQTemplate, NewsStatsService newsStatsService) {
        this.rocketMQTemplate = rocketMQTemplate;
        this.newsStatsService = newsStatsService;
    }

    public void sendNormalMessage(String topic, String tag, News content)
    {
        sendNormalMessage(topic, tag, content, false);
    }

    /**
     * 发送普通消息
     * 适用场景：不需要顺序保证、事务处理的普通业务通知
     * 例如：订单创建通知、库存更新通知等
     *
     * @param topic 消息主题
     * @param tag   消息标签（用于消息过滤）
     * @param content 消息内容
     */
    public void sendNormalMessage(String topic, String tag, News content, boolean async) {
        // 构建完整目的地（主题:标签）
        String destination = topic + ":" + tag;

        // 创建Spring Messaging Message对象
        Message<News> message = MessageBuilder.withPayload(content)
                // 设置业务键（用于追踪）
                .setHeader("KEYS", "news_id_" + content.getId())
                // 设置其他自定义属性
                .setHeader("business_type", "news_created")
                .build();

        if (async){
            asyncSend(destination, message);
        }else {
            syncSend(destination, message);
        }

    }

    private void syncSend(String destination, Message<News> message){
        // 发送消息（同步方式）
        SendResult sendResult = rocketMQTemplate.syncSend(destination, message);
        if(sendResult.getSendStatus() == SendStatus.SEND_OK) {
            log.info("普通消息发送成功: {}, 消息ID: {}", destination, sendResult.getMsgId());
        }else {
            log.error("普通消息发送失败: {}, 失败状态: {}", destination, sendResult.getSendStatus());
        }
    }

    private void asyncSend(String destination, Message<News> message) {
        // 异步发送示例（适合高吞吐场景）
        rocketMQTemplate.asyncSend(destination, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("异步发送成功: " + sendResult.getMsgId());
            }
            @Override
            public void onException(Throwable throwable) {
                System.err.println("异步发送失败: " + throwable.getMessage());
            }
        });
    }

    /**
     * 发送延时消息
     * 适用场景：需要延迟处理的业务场景
     * 例如：订单支付超时、库存冻结时长、定时任务等
     *
     * @param topic 延时消息主题
     * @param tag   延时消息标签（用于消息过滤）
     */
    public void sendDelayMessage(String topic, String tag, News content, long delay) {
        // 构建完整目的地（主题:标签）
        String destination = topic + ":" + tag;

        // 创建Spring Messaging Message对象
        Message<News> message = MessageBuilder.withPayload(content)
                // 设置业务键（用于追踪）
                .setHeader("KEYS", "news_id_" + content.getId())
                // 设置其他自定义属性
                .setHeader("business_type", "news_created")
                .build();

        SendResult result;
        // 发送延迟消息毫秒级（需要Broker配置，enableScheduleTimeMessage = true）
        //result = rocketMQTemplate.syncSendDelayTimeMills(destination, message, delay);
        // 发送延迟消息秒级（最大延迟2小时）
        result = rocketMQTemplate.syncSendDelayTimeSeconds(destination, message, delay);
        // 延迟消息指定时间戳
        // result = rocketMQTemplate.syncSendDeliverTimeMills(destination, message, delay);

        if(result.getSendStatus() == SendStatus.SEND_OK) {
            log.info("延迟消息发送成功: {}, 消息ID: {}", destination, result.getMsgId());
        }else {
            log.error("延迟消息发送失败: {}, 失败状态: {}", destination, result.getSendStatus());
        }
    }

    /**
     * 获取最热门新闻
     *
     * @param topN 最多返回的条数
     * @return 最热门新闻列表
     */
    @Override
    public List<TopNewsDTO> getTopNews(int topN) {
        return newsStatsService.getTopViewedNews(topN);
    }

    @Override
    public News getOneNewsWith(Long id) {
        // 1.设置缓存
        // News oneNews = getNewsByIdWithCaffeine(id);

        // 2.从缓存中获取
        // 类内直接调用getNewsByIdWithCache方法，缓存功能会失效，因为AOP代理对象无法获取缓存
        // News oneNews = getNewsByIdWithCache(id);
        // 使用AOP代理对象再获取缓存内容，缓存功能正常（配合@EnableAspectJAutoProxy(exposeProxy = true) ）
        NewsServiceImpl proxy = (NewsServiceImpl) AopContext.currentProxy();
        News oneNews = proxy.getNewsByIdWithCache(id);


        // 2.增加浏览量
        newsStatsService.incrementViews(id);

        //  TODO: 3.记录用户行为数据（浏览文章，点击轨迹等）

        return oneNews;
    }// AOP

    @Override
    public News getNewsByIdNewsWith(Long aLong) {
        return null;
    }


    // 缓存方法结果，缓存名为 "news"，键为 id
    @Cacheable(value = "news", key = "#id", cacheManager = "redis")
    public News getNewsByIdWithCache(Long id) {
        log.info("从数据库中获取新闻信息：{}", id);
        return getById(id);
    }

    /**-----手动缓存数据-----**/
    // 初始化缓存：以 news:id 为键，缓存新闻对象
    private final Cache<String, News> newsCache = Caffeine.newBuilder()
            .expireAfterWrite(10, TimeUnit.MINUTES) // 设置缓存过期时间
            .maximumSize(100) // 设置最大缓存条目数
            .build();
    public News getNewsByIdWithCaffeine(Long id){
        log.info("使用Caffeine缓存新闻：{}", id);
        String cacheKey = "news:" + id; // 构造缓存键

        // 1. 尝试从缓存中获取新闻
        News news = newsCache.getIfPresent(cacheKey);
        if (news != null) {
            log.info("从缓存中获取新闻信息：{}", id);
            return news; // 缓存命中，直接返回
        }

        // 2. 缓存未命中，查询数据库
        log.info("查询数据库获得指定ID的新闻");
        news = getById(id);

        // 3. 判断是否需要缓存
        // 这里可以根据新闻的某些属性（如发布时间、类型等）进行判断
        // 示例条件：如果新闻内容长度大于 100 字，则缓存
        if (news != null && news.getContent() != null) {
            log.info("将新闻信息加入缓存：{}", id);
            newsCache.put(cacheKey, news); // 将新闻加入缓存
        }

        return news;
    }

}
